package com.zscat.mallplus.qsg.impl;

import com.alibaba.fastjson.JSONArray;
import com.alibaba.fastjson.JSONObject;
import com.zscat.mallplus.qsg.BilliardWeighService;
import lombok.extern.slf4j.Slf4j;
import org.eclipse.paho.client.mqttv3.MqttException;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;

import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.FutureTask;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

@Service
@Slf4j
public class BilliardWeighServiceImpl implements BilliardWeighService {

    @Autowired
    private MqttConfigService mqttConfigService;

    private final ThreadPoolExecutor threadPoolExecutor =
            new ThreadPoolExecutor(3, 5, 30, TimeUnit.SECONDS, new ArrayBlockingQueue<>(100), runnable -> {
                Thread thread = new Thread(runnable, "mqtt thread");
                return thread;
            });

    @Override
    public Integer send(String deviceCode) throws Exception {
        FutureTask<Integer> futureTask = sendTask(deviceCode);
        return futureTask.get();
    }

    @Override
    public void open(String deviceCode) throws MqttException {
        JSONObject dataInfo = new JSONObject();
        JSONObject headerInfo = new JSONObject();
        JSONObject bodyInfo = new JSONObject();
        headerInfo.put("Code","1004");
        headerInfo.put("Time", System.currentTimeMillis());
        headerInfo.put("Sign","lvzaina");
        dataInfo.put("Header",headerInfo);
        bodyInfo.put("MsgId",1683557943);
        bodyInfo.put("Cmd","Relay");
        JSONArray relays = new JSONArray();
        JSONObject relayOne = new JSONObject();
        relayOne.put("Index",2);
        relayOne.put("Delay",4294967295L);
        relays.add(relayOne);
        JSONObject relayTwo = new JSONObject();
        relayTwo.put("Index",3);
        relayTwo.put("Delay",4294967295L);
        relays.add(relayTwo);
        bodyInfo.put("Relays",relays);
        dataInfo.put("Body",bodyInfo);
        mqttConfigService.publish("topic/receive/"+ deviceCode, dataInfo.toJSONString());
    }

    @Override
    public void close(String deviceCode) throws MqttException {
        JSONObject dataInfo = new JSONObject();
        JSONObject headerInfo = new JSONObject();
        JSONObject bodyInfo = new JSONObject();
        headerInfo.put("Code","1004");
        headerInfo.put("Time", System.currentTimeMillis());
        headerInfo.put("Sign","lvzaina");
        dataInfo.put("Header",headerInfo);
        bodyInfo.put("MsgId",1683557943);
        bodyInfo.put("Cmd","Relay");
        JSONArray relays = new JSONArray();
        JSONObject relayOne = new JSONObject();
        relayOne.put("Index",2);
        relayOne.put("Delay",0);
        relays.add(relayOne);
        JSONObject relayTwo = new JSONObject();
        relayTwo.put("Index",3);
        relayTwo.put("Delay",0);
        relays.add(relayTwo);
        bodyInfo.put("Relays",relays);
        dataInfo.put("Body",bodyInfo);
        mqttConfigService.publish("topic/receive/"+ deviceCode, dataInfo.toJSONString());
    }

    private FutureTask<Integer> sendTask(String deviceCode) {
        FutureTask<Integer> futureTask = new FutureTask<>(new GetMqttDataValueCallable(deviceCode,mqttConfigService));
        threadPoolExecutor.execute(futureTask);
        //阻塞线程
        return futureTask;
    }
}
